package api.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Json;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;

/**
 *  从kafka读取数据  (connect方法)
 *  输出到 MySQL JDBC  (SQL-DDL)
 */
public class TableTest6_jdbc1 {

    public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000L);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String topicFrom = "myTest";
        String kafkaServer = "192.168.36.130:9092,192.168.36.129:9092,192.168.36.128:9092";

        String db_url = "jdbc:mysql://192.168.36.129:3306/test?useUnicode=true&characterEncoding=UTF-8&useSSL=false";
        String db_driver = "com.mysql.cj.jdbc.Driver";
        String db_username = "root";
        String db_password = "root";
        String db_table = "flink_test";

        // 2. 连接Kafka，读取数据
        Kafka kafka = new Kafka().version("universal").topic(topicFrom).property("bootstrap.servers", kafkaServer);
        Schema schema = new Schema()
            .field("id", DataTypes.STRING())
            .field("timestamp", DataTypes.BIGINT())
            .field("temp", DataTypes.DOUBLE());
        tableEnv.connect(kafka)
                .withFormat(new Json())
                .withSchema(schema)
                .inAppendMode()
                .createTemporaryTable("inputTable");
        // 3. 查询转换
        Table table = tableEnv.sqlQuery("select id, temp from inputTable");
        // 打印查看
        tableEnv.toRetractStream(table, Row.class).print("data:");

        String sinkDDL = "create table jdbcOutputTable (" +
                        " id varchar(255) not null, " +
                        " temp double " +
                        ") with (" +
                        " 'connector.type' = 'jdbc', " +
                        " 'connector.url' = '" + db_url + "', " +
                        " 'connector.table' = '" + db_table + "', " +
                        " 'connector.driver' = '" + db_driver + "', " +
                        " 'connector.username' = '" + db_username + "', " +
                        " 'connector.password' = '" + db_password + "' )";
        tableEnv.executeSql(sinkDDL);
        table.executeInsert("jdbcOutputTable");

    }

}
